{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data Processing using PySpark\n",
    "\n",
    "    \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark=SparkSession.builder.appName('data_processing').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "#creating a customer dataframe by declaring the schema and passing values\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql.types import *\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create custom Dataframe by declaring the schema "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "schema=StructType().add(\"user_id\",\"string\").add(\"country\",\"string\").add(\"browser\", \"string\").add(\"OS\",'string').add(\"age\", \"integer\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "#pass the values\n",
    "df=spark.createDataFrame([(\"A203\",'India',\"Chrome\",\"WIN\",33),(\"A201\",'China',\"Safari\",\"MacOS\",35),(\"A205\",'UK',\"Mozilla\",\"Linux\",25)],schema=schema)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- user_id: string (nullable = true)\n",
      " |-- country: string (nullable = true)\n",
      " |-- browser: string (nullable = true)\n",
      " |-- OS: string (nullable = true)\n",
      " |-- age: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+-------+-----+---+\n",
      "|user_id|country|browser|   OS|age|\n",
      "+-------+-------+-------+-----+---+\n",
      "|   A203|  India| Chrome|  WIN| 33|\n",
      "|   A201|  China| Safari|MacOS| 35|\n",
      "|   A205|     UK|Mozilla|Linux| 25|\n",
      "+-------+-------+-------+-----+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create a new dataframe with nukll values \n",
    "df_na=spark.createDataFrame([(\"A203\",None,\"Chrome\",\"WIN\",33),(\"A201\",'China',None,\"MacOS\",35),(\"A205\",'UK',\"Mozilla\",\"Linux\",25)],schema=schema)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+-------+-----+---+\n",
      "|user_id|country|browser|   OS|age|\n",
      "+-------+-------+-------+-----+---+\n",
      "|   A203|   null| Chrome|  WIN| 33|\n",
      "|   A201|  China|   null|MacOS| 35|\n",
      "|   A205|     UK|Mozilla|Linux| 25|\n",
      "+-------+-------+-------+-----+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_na.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+-------+-----+---+\n",
      "|user_id|country|browser|   OS|age|\n",
      "+-------+-------+-------+-----+---+\n",
      "|   A203|      0| Chrome|  WIN| 33|\n",
      "|   A201|  China|      0|MacOS| 35|\n",
      "|   A205|     UK|Mozilla|Linux| 25|\n",
      "+-------+-------+-------+-----+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#fill all null values with 0\n",
    "df_na.fillna('0').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 232,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+-------+-----+---+\n",
      "|user_id|country|browser|   OS|age|\n",
      "+-------+-------+-------+-----+---+\n",
      "|   A203|    USA| Chrome|  WIN| 33|\n",
      "|   A201|  China| Safari|MacOS| 35|\n",
      "|   A205|     UK|Mozilla|Linux| 25|\n",
      "+-------+-------+-------+-----+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#fill null values with specific value\n",
    "df_na.fillna( { 'country':'USA', 'browser':'Safari' } ).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 233,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+-------+-----+---+\n",
      "|user_id|country|browser|   OS|age|\n",
      "+-------+-------+-------+-----+---+\n",
      "|   A205|     UK|Mozilla|Linux| 25|\n",
      "+-------+-------+-------+-----+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Return new df omitting rows with null values\n",
    "df_na.na.drop().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 234,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+-------+-----+---+\n",
      "|user_id|country|browser|   OS|age|\n",
      "+-------+-------+-------+-----+---+\n",
      "|   A201|  China|   null|MacOS| 35|\n",
      "|   A205|     UK|Mozilla|Linux| 25|\n",
      "+-------+-------+-------+-----+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_na.na.drop(subset='country').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 235,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+-------------+-----+---+\n",
      "|user_id|country|      browser|   OS|age|\n",
      "+-------+-------+-------------+-----+---+\n",
      "|   A203|   null|Google Chrome|  WIN| 33|\n",
      "|   A201|  China|         null|MacOS| 35|\n",
      "|   A205|     UK|      Mozilla|Linux| 25|\n",
      "+-------+-------+-------------+-----+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_na.replace(\"Chrome\",\"Google Chrome\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 236,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+-----+---+\n",
      "|country|browser|   OS|age|\n",
      "+-------+-------+-----+---+\n",
      "|  India| Chrome|  WIN| 33|\n",
      "|  China| Safari|MacOS| 35|\n",
      "|     UK|Mozilla|Linux| 25|\n",
      "+-------+-------+-----+---+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#deleting column \n",
    "df.drop('user_id').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#read files as Dataframe "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 240,
   "metadata": {},
   "outputs": [],
   "source": [
    "df=spark.read.csv(\"customer_data.csv\",header=True,inferSchema=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2000"
      ]
     },
     "execution_count": 74,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "7"
      ]
     },
     "execution_count": 75,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "len(df.columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Customer_subtype: string (nullable = true)\n",
      " |-- Number_of_houses: integer (nullable = true)\n",
      " |-- Avg_size_household: integer (nullable = true)\n",
      " |-- Avg_age: string (nullable = true)\n",
      " |-- Customer_main_type: string (nullable = true)\n",
      " |-- Avg_Salary: integer (nullable = true)\n",
      " |-- label: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+\n",
      "|    Customer_subtype|Number_of_houses|Avg_size_household|    Avg_age|  Customer_main_type|Avg_Salary|label|\n",
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+\n",
      "|Lower class large...|               1|                 3|30-40 years|Family with grown...|     44905|    0|\n",
      "|Mixed small town ...|               1|                 2|30-40 years|Family with grown...|     37575|    0|\n",
      "|Mixed small town ...|               1|                 2|30-40 years|Family with grown...|     27915|    0|\n",
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Customer_subtype',\n",
       " 'Number_of_houses',\n",
       " 'Avg_size_household',\n",
       " 'Avg_age',\n",
       " 'Customer_main_type',\n",
       " 'Avg_Salary',\n",
       " 'label']"
      ]
     },
     "execution_count": 68,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+--------------------+------------------+------------------+-----------+--------------------+-----------------+------------------+\n",
      "|summary|    Customer_subtype|  Number_of_houses|Avg_size_household|    Avg_age|  Customer_main_type|       Avg_Salary|             label|\n",
      "+-------+--------------------+------------------+------------------+-----------+--------------------+-----------------+------------------+\n",
      "|  count|                2000|              2000|              2000|       2000|                2000|             2000|              2000|\n",
      "|   mean|                null|            1.1075|            2.6895|       null|                null|     1616908.0835|            0.0605|\n",
      "| stddev|                null|0.3873225521186316|0.7914562220841646|       null|                null|6822647.757312146|0.2384705099001677|\n",
      "|    min|Affluent senior a...|                 1|                 1|20-30 years|      Average Family|             1361|                 0|\n",
      "|    25%|                null|                 1|                 2|       null|                null|            20315|                 0|\n",
      "|    50%|                null|                 1|                 3|       null|                null|            31421|                 0|\n",
      "|    75%|                null|                 1|                 3|       null|                null|            42949|                 0|\n",
      "|    max| Young, low educated|                10|                 5|70-80 years|Successful hedonists|         48919896|                 1|\n",
      "+-------+--------------------+------------------+------------------+-----------+--------------------+-----------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.summary().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Subset "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------+\n",
      "|    Customer_subtype|Avg_Salary|\n",
      "+--------------------+----------+\n",
      "|Lower class large...|     44905|\n",
      "|Mixed small town ...|     37575|\n",
      "|Mixed small town ...|     27915|\n",
      "|Modern, complete ...|     19504|\n",
      "|  Large family farms|     34943|\n",
      "|    Young and rising|     13064|\n",
      "|Large religious f...|     29090|\n",
      "|Lower class large...|      6895|\n",
      "|Lower class large...|     35497|\n",
      "|     Family starters|     30800|\n",
      "|       Stable family|     39157|\n",
      "|Modern, complete ...|     40839|\n",
      "|Lower class large...|     30008|\n",
      "|        Mixed rurals|     37209|\n",
      "|    Young and rising|     45361|\n",
      "|Lower class large...|     45650|\n",
      "|Traditional families|     18982|\n",
      "|Mixed apartment d...|     30093|\n",
      "|Young all america...|     27097|\n",
      "|Low income catholics|     23511|\n",
      "+--------------------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(['Customer_subtype','Avg_Salary']).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "128"
      ]
     },
     "execution_count": 96,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.filter(df['Avg_Salary'] > 1000000).count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+\n",
      "|    Customer_subtype|Number_of_houses|Avg_size_household|    Avg_age|  Customer_main_type|Avg_Salary|label|\n",
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+\n",
      "| High status seniors|               1|                 3|40-50 years|Successful hedonists|   4670288|    0|\n",
      "| High status seniors|               1|                 3|50-60 years|Successful hedonists|   9561873|    0|\n",
      "| High status seniors|               1|                 2|40-50 years|Successful hedonists|  18687005|    0|\n",
      "| High status seniors|               1|                 2|40-50 years|Successful hedonists|  24139960|    0|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|   6718606|    0|\n",
      "|High Income, expe...|               1|                 3|40-50 years|Successful hedonists|  19347139|    0|\n",
      "|High Income, expe...|               1|                 3|40-50 years|Successful hedonists|   5243902|    0|\n",
      "| High status seniors|               2|                 3|40-50 years|Successful hedonists|   6138618|    0|\n",
      "|High Income, expe...|               1|                 3|50-60 years|Successful hedonists|  24930053|    0|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|  12545905|    1|\n",
      "|High Income, expe...|               1|                 3|40-50 years|Successful hedonists|  29976435|    0|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|  24639614|    0|\n",
      "| High status seniors|               1|                 2|40-50 years|Successful hedonists|  16073966|    0|\n",
      "|High Income, expe...|               1|                 4|40-50 years|Successful hedonists|  35032441|    1|\n",
      "|High Income, expe...|               1|                 2|50-60 years|Successful hedonists|   8354410|    0|\n",
      "| High status seniors|               1|                 1|60-70 years|Successful hedonists|  20241068|    0|\n",
      "| High status seniors|               1|                 1|50-60 years|Successful hedonists|  45592572|    0|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|  10289449|    0|\n",
      "|High Income, expe...|               1|                 2|50-60 years|Successful hedonists|   5586401|    0|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|  41699271|    0|\n",
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(df['Avg_Salary'] > 1000000).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 106,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+\n",
      "|    Customer_subtype|Number_of_houses|Avg_size_household|    Avg_age|  Customer_main_type|Avg_Salary|label|\n",
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+\n",
      "|Affluent senior a...|               3|                 2|50-60 years|Successful hedonists|    596723|    0|\n",
      "|Affluent senior a...|               3|                 2|50-60 years|Successful hedonists|    944444|    0|\n",
      "|Affluent senior a...|               3|                 2|50-60 years|Successful hedonists|    788477|    0|\n",
      "|Affluent senior a...|               3|                 2|50-60 years|Successful hedonists|    994077|    0|\n",
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(df['Avg_Salary'] > 500000).filter(df['Number_of_houses'] > 2).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 107,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+\n",
      "|    Customer_subtype|Number_of_houses|Avg_size_household|    Avg_age|  Customer_main_type|Avg_Salary|label|\n",
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+\n",
      "|Affluent senior a...|               3|                 2|50-60 years|Successful hedonists|    596723|    0|\n",
      "|Affluent senior a...|               3|                 2|50-60 years|Successful hedonists|    944444|    0|\n",
      "|Affluent senior a...|               3|                 2|50-60 years|Successful hedonists|    788477|    0|\n",
      "|Affluent senior a...|               3|                 2|50-60 years|Successful hedonists|    994077|    0|\n",
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.where((df['Avg_Salary'] > 500000) & (df['Number_of_houses'] > 2)).show()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Aggregations "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 110,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----+\n",
      "|    Customer_subtype|count|\n",
      "+--------------------+-----+\n",
      "|Large family, emp...|   56|\n",
      "|Religious elderly...|   47|\n",
      "|Large religious f...|  107|\n",
      "|Modern, complete ...|   93|\n",
      "|    Village families|   68|\n",
      "|Young all america...|   62|\n",
      "|Young urban have-...|    4|\n",
      "|Young seniors in ...|   22|\n",
      "|Fresh masters in ...|    2|\n",
      "|High Income, expe...|   52|\n",
      "|Lower class large...|  288|\n",
      "| Residential elderly|    6|\n",
      "|Senior cosmopolitans|    1|\n",
      "|        Mixed rurals|   67|\n",
      "|Career and childcare|   33|\n",
      "|Low income catholics|   72|\n",
      "|Mixed apartment d...|   34|\n",
      "|Seniors in apartm...|   17|\n",
      "|Middle class fami...|  122|\n",
      "|Traditional families|  129|\n",
      "+--------------------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('Customer_subtype').count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 112,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " *** Aggregation for  Customer_subtype ***\n",
      "+------------------------------------------+-----+\n",
      "|Customer_subtype                          |count|\n",
      "+------------------------------------------+-----+\n",
      "|Lower class large families                |288  |\n",
      "|Traditional families                      |129  |\n",
      "|Middle class families                     |122  |\n",
      "|Large religious families                  |107  |\n",
      "|Modern, complete families                 |93   |\n",
      "|Couples with teens 'Married with children'|83   |\n",
      "|Young and rising                          |78   |\n",
      "|High status seniors                       |76   |\n",
      "|Low income catholics                      |72   |\n",
      "|Mixed seniors                             |71   |\n",
      "|Village families                          |68   |\n",
      "|Mixed rurals                              |67   |\n",
      "|Young all american family                 |62   |\n",
      "|Stable family                             |62   |\n",
      "|Young, low educated                       |56   |\n",
      "|Large family, employed child              |56   |\n",
      "|Family starters                           |55   |\n",
      "|High Income, expensive child              |52   |\n",
      "|Mixed small town dwellers                 |47   |\n",
      "|Religious elderly singles                 |47   |\n",
      "+------------------------------------------+-----+\n",
      "only showing top 20 rows\n",
      "\n",
      " *** Aggregation for  Number_of_houses ***\n",
      "+----------------+-----+\n",
      "|Number_of_houses|count|\n",
      "+----------------+-----+\n",
      "|1               |1808 |\n",
      "|2               |178  |\n",
      "|3               |12   |\n",
      "|5               |1    |\n",
      "|10              |1    |\n",
      "+----------------+-----+\n",
      "\n",
      " *** Aggregation for  Avg_size_household ***\n",
      "+------------------+-----+\n",
      "|Avg_size_household|count|\n",
      "+------------------+-----+\n",
      "|3                 |900  |\n",
      "|2                 |730  |\n",
      "|4                 |255  |\n",
      "|1                 |94   |\n",
      "|5                 |21   |\n",
      "+------------------+-----+\n",
      "\n",
      " *** Aggregation for  Avg_age ***\n",
      "+-----------+-----+\n",
      "|Avg_age    |count|\n",
      "+-----------+-----+\n",
      "|40-50 years|1028 |\n",
      "|30-40 years|496  |\n",
      "|50-60 years|373  |\n",
      "|60-70 years|64   |\n",
      "|20-30 years|31   |\n",
      "|70-80 years|8    |\n",
      "+-----------+-----+\n",
      "\n",
      " *** Aggregation for  Customer_main_type ***\n",
      "+---------------------+-----+\n",
      "|Customer_main_type   |count|\n",
      "+---------------------+-----+\n",
      "|Family with grown ups|542  |\n",
      "|Average Family       |308  |\n",
      "|Conservative families|236  |\n",
      "|Retired and Religious|202  |\n",
      "|Successful hedonists |194  |\n",
      "|Living well          |178  |\n",
      "|Driven Growers       |172  |\n",
      "|Farmers              |93   |\n",
      "|Cruising Seniors     |60   |\n",
      "|Career Loners        |15   |\n",
      "+---------------------+-----+\n",
      "\n",
      " *** Aggregation for  label ***\n",
      "+-----+-----+\n",
      "|label|count|\n",
      "+-----+-----+\n",
      "|0    |1879 |\n",
      "|1    |121  |\n",
      "+-----+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "for col in df.columns:\n",
    "    if col !='Avg_Salary':\n",
    "        print(f\" *** Aggregation for  {col} ***\")\n",
    "        df.groupBy(col).count().orderBy('count',ascending=False).show(truncate=False)\n",
    "\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 116,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+---------------+\n",
      "|  Customer_main_type|min(Avg_Salary)|\n",
      "+--------------------+---------------+\n",
      "|             Farmers|          10469|\n",
      "|       Career Loners|          13246|\n",
      "|Retired and Relig...|           1361|\n",
      "|Successful hedonists|          12705|\n",
      "|         Living well|          10418|\n",
      "|      Average Family|          10506|\n",
      "|    Cruising Seniors|          10100|\n",
      "|Conservative fami...|          10179|\n",
      "|      Driven Growers|          10257|\n",
      "|Family with grown...|           1502|\n",
      "+--------------------+---------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('Customer_main_type').agg(F.min('Avg_Salary')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 117,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+---------------+\n",
      "|  Customer_main_type|max(Avg_Salary)|\n",
      "+--------------------+---------------+\n",
      "|             Farmers|          49965|\n",
      "|       Career Loners|          49903|\n",
      "|Retired and Relig...|          49564|\n",
      "|Successful hedonists|       48919896|\n",
      "|         Living well|          49816|\n",
      "|      Average Family|         991838|\n",
      "|    Cruising Seniors|          49526|\n",
      "|Conservative fami...|          49965|\n",
      "|      Driven Growers|          49932|\n",
      "|Family with grown...|          49901|\n",
      "+--------------------+---------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('Customer_main_type').agg(F.max('Avg_Salary')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+---------------+\n",
      "|  Customer_main_type|sum(Avg_Salary)|\n",
      "+--------------------+---------------+\n",
      "|             Farmers|        2809468|\n",
      "|       Career Loners|         484089|\n",
      "|Retired and Relig...|        5522439|\n",
      "|Successful hedonists|     3158111161|\n",
      "|         Living well|        5552540|\n",
      "|      Average Family|       32111040|\n",
      "|    Cruising Seniors|        1732220|\n",
      "|Conservative fami...|        6963043|\n",
      "|      Driven Growers|        5292275|\n",
      "|Family with grown...|       15237892|\n",
      "+--------------------+---------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('Customer_main_type').agg(F.sum('Avg_Salary')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 119,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------------------+\n",
      "|  Customer_main_type|     avg(Avg_Salary)|\n",
      "+--------------------+--------------------+\n",
      "|             Farmers|  30209.333333333332|\n",
      "|       Career Loners|             32272.6|\n",
      "|Retired and Relig...|   27338.80693069307|\n",
      "|Successful hedonists|1.6278923510309279E7|\n",
      "|         Living well|  31194.044943820223|\n",
      "|      Average Family|  104256.62337662338|\n",
      "|    Cruising Seniors|  28870.333333333332|\n",
      "|Conservative fami...|  29504.419491525423|\n",
      "|      Driven Growers|   30769.04069767442|\n",
      "|Family with grown...|  28114.191881918818|\n",
      "+--------------------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('Customer_main_type').agg(F.mean('Avg_Salary')).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Sorting "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 169,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+\n",
      "|    Customer_subtype|Number_of_houses|Avg_size_household|    Avg_age|  Customer_main_type|Avg_Salary|label|\n",
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+\n",
      "| High status seniors|               1|                 2|60-70 years|Successful hedonists|  48919896|    0|\n",
      "|High Income, expe...|               1|                 2|50-60 years|Successful hedonists|  48177970|    0|\n",
      "|High Income, expe...|               1|                 2|50-60 years|Successful hedonists|  48069548|    1|\n",
      "|High Income, expe...|               1|                 3|40-50 years|Successful hedonists|  46911924|    0|\n",
      "| High status seniors|               1|                 3|40-50 years|Successful hedonists|  46614009|    0|\n",
      "|High Income, expe...|               1|                 3|30-40 years|Successful hedonists|  45952441|    0|\n",
      "|High Income, expe...|               1|                 3|40-50 years|Successful hedonists|  45864609|    1|\n",
      "| High status seniors|               1|                 1|50-60 years|Successful hedonists|  45592572|    0|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|  45170899|    0|\n",
      "|High Income, expe...|               1|                 3|50-60 years|Successful hedonists|  44843830|    0|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|  43230349|    0|\n",
      "| High status seniors|               1|                 5|40-50 years|Successful hedonists|  43181830|    0|\n",
      "| High status seniors|               1|                 1|30-40 years|Successful hedonists|  42631926|    0|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|  41919020|    0|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|  41699271|    0|\n",
      "| High status seniors|               1|                 2|40-50 years|Successful hedonists|  41398953|    0|\n",
      "|High Income, expe...|               1|                 3|40-50 years|Successful hedonists|  41269615|    1|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|  41192397|    1|\n",
      "| High status seniors|               1|                 2|40-50 years|Successful hedonists|  40564335|    0|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|  40453887|    0|\n",
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.sort(\"Avg_salary\", ascending=False).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 126,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------------------------+--------------------+\n",
      "|Customer_subtype                          |mean_salary         |\n",
      "+------------------------------------------+--------------------+\n",
      "|High status seniors                       |2.507677857894737E7 |\n",
      "|High Income, expensive child              |2.3839817807692308E7|\n",
      "|Affluent young families                   |662068.7777777778   |\n",
      "|Affluent senior apartments                |653638.8235294118   |\n",
      "|Senior cosmopolitans                      |49903.0             |\n",
      "|Students in apartments                    |35532.142857142855  |\n",
      "|Large family farms                        |33135.61538461538   |\n",
      "|Young, low educated                       |33072.21428571428   |\n",
      "|Large family, employed child              |32867.857142857145  |\n",
      "|Suburban youth                            |32558.0             |\n",
      "|Village families                          |32449.470588235294  |\n",
      "|Middle class families                     |31579.385245901638  |\n",
      "|Modern, complete families                 |31576.0             |\n",
      "|Etnically diverse                         |31572.0             |\n",
      "|Young and rising                          |30795.897435897437  |\n",
      "|Mixed seniors                             |30759.267605633802  |\n",
      "|Very Important Provincials                |30548.40625         |\n",
      "|Religious elderly singles                 |30540.59574468085   |\n",
      "|Family starters                           |30376.2             |\n",
      "|Career and childcare                      |30110.939393939392  |\n",
      "|Young seniors in the city                 |30105.136363636364  |\n",
      "|Seniors in apartments                     |30090.882352941175  |\n",
      "|Large religious families                  |29652.196261682242  |\n",
      "|Stable family                             |29619.032258064515  |\n",
      "|Mixed apartment dwellers                  |29587.647058823528  |\n",
      "|Young all american family                 |29563.3064516129    |\n",
      "|Porchless seniors: no front yard          |29509.827586206895  |\n",
      "|Traditional families                      |29381.84496124031   |\n",
      "|Mixed rurals                              |29073.761194029852  |\n",
      "|Mixed small town dwellers                 |28982.106382978724  |\n",
      "|Residential elderly                       |28866.166666666668  |\n",
      "|Couples with teens 'Married with children'|28155.807228915663  |\n",
      "|Fresh masters in the city                 |27645.0             |\n",
      "|Dinki's (double income no kids)           |26231.117647058825  |\n",
      "|Lower class large families                |26012.628472222223  |\n",
      "|Young urban have-nots                     |25751.0             |\n",
      "|Own home elderly                          |25677.666666666668  |\n",
      "|Single youth                              |24403.25            |\n",
      "|Low income catholics                      |21713.777777777777  |\n",
      "+------------------------------------------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('Customer_subtype').agg(F.avg('Avg_Salary').alias('mean_salary')).orderBy('mean_salary',ascending=False).show(50,False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 123,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------+\n",
      "|    Customer_subtype|max_salary|\n",
      "+--------------------+----------+\n",
      "| High status seniors|  48919896|\n",
      "|High Income, expe...|  48177970|\n",
      "|Affluent senior a...|    994077|\n",
      "|Affluent young fa...|    991838|\n",
      "|  Large family farms|     49965|\n",
      "|Traditional families|     49965|\n",
      "|Middle class fami...|     49932|\n",
      "|Senior cosmopolitans|     49903|\n",
      "|Mixed small town ...|     49901|\n",
      "|Lower class large...|     49899|\n",
      "|       Mixed seniors|     49876|\n",
      "|    Young and rising|     49816|\n",
      "|        Mixed rurals|     49785|\n",
      "|Modern, complete ...|     49729|\n",
      "| Young, low educated|     49626|\n",
      "|Mixed apartment d...|     49621|\n",
      "|     Family starters|     49602|\n",
      "|    Village families|     49575|\n",
      "|Religious elderly...|     49564|\n",
      "|       Stable family|     49548|\n",
      "+--------------------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('Customer_subtype').agg(F.max('Avg_Salary').alias('max_salary')).orderBy('max_salary',ascending=False).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### Collect "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 147,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----------------------------+\n",
      "|    Customer_subtype|collect_set(Number_of_houses)|\n",
      "+--------------------+-----------------------------+\n",
      "|Large family, emp...|                       [1, 2]|\n",
      "|Religious elderly...|                       [1, 2]|\n",
      "|Large religious f...|                       [1, 2]|\n",
      "|Modern, complete ...|                       [1, 2]|\n",
      "|    Village families|                       [1, 2]|\n",
      "|Young all america...|                       [1, 2]|\n",
      "|Young urban have-...|                       [1, 2]|\n",
      "|Young seniors in ...|                    [1, 2, 3]|\n",
      "|Fresh masters in ...|                          [1]|\n",
      "|High Income, expe...|                          [1]|\n",
      "|Lower class large...|                       [1, 2]|\n",
      "| Residential elderly|                    [1, 2, 3]|\n",
      "|Senior cosmopolitans|                          [3]|\n",
      "|        Mixed rurals|                          [1]|\n",
      "|Career and childcare|                       [1, 2]|\n",
      "|Low income catholics|                          [1]|\n",
      "|Mixed apartment d...|                    [1, 2, 3]|\n",
      "|Seniors in apartm...|                       [1, 2]|\n",
      "|Middle class fami...|                       [1, 2]|\n",
      "|Traditional families|                       [1, 2]|\n",
      "+--------------------+-----------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Collect _set \n",
    "df.groupby(\"Customer_subtype\").agg(F.collect_set(\"Number_of_houses\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 149,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------------------------+\n",
      "|    Customer_subtype|collect_list(Number_of_houses)|\n",
      "+--------------------+------------------------------+\n",
      "|Large family, emp...|          [2, 1, 2, 1, 2, 1...|\n",
      "|Religious elderly...|          [1, 1, 1, 1, 1, 1...|\n",
      "|Large religious f...|          [2, 1, 1, 2, 1, 1...|\n",
      "|Modern, complete ...|          [1, 1, 2, 1, 1, 1...|\n",
      "|    Village families|          [1, 1, 1, 1, 1, 1...|\n",
      "|Young all america...|          [1, 1, 2, 2, 1, 1...|\n",
      "|Young urban have-...|                  [1, 2, 1, 1]|\n",
      "|Young seniors in ...|          [1, 1, 1, 1, 1, 2...|\n",
      "|Fresh masters in ...|                        [1, 1]|\n",
      "|High Income, expe...|          [1, 1, 1, 1, 1, 1...|\n",
      "|Lower class large...|          [1, 1, 1, 1, 1, 1...|\n",
      "| Residential elderly|            [3, 1, 1, 3, 2, 1]|\n",
      "|Senior cosmopolitans|                           [3]|\n",
      "|        Mixed rurals|          [1, 1, 1, 1, 1, 1...|\n",
      "|Career and childcare|          [2, 1, 1, 1, 1, 1...|\n",
      "|Low income catholics|          [1, 1, 1, 1, 1, 1...|\n",
      "|Mixed apartment d...|          [2, 3, 1, 1, 1, 1...|\n",
      "|Seniors in apartm...|          [2, 2, 2, 2, 1, 2...|\n",
      "|Middle class fami...|          [1, 1, 2, 1, 1, 1...|\n",
      "|Traditional families|          [1, 1, 1, 1, 1, 1...|\n",
      "+--------------------+------------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#collect list \n",
    "df.groupby(\"Customer_subtype\").agg(F.collect_list(\"Number_of_houses\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 241,
   "metadata": {},
   "outputs": [],
   "source": [
    "#creating a new column with constant value\n",
    "\n",
    "df=df.withColumn('constant',F.lit('finance'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 242,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+--------+\n",
      "|    Customer_subtype|constant|\n",
      "+--------------------+--------+\n",
      "|Lower class large...| finance|\n",
      "|Mixed small town ...| finance|\n",
      "|Mixed small town ...| finance|\n",
      "|Modern, complete ...| finance|\n",
      "|  Large family farms| finance|\n",
      "|    Young and rising| finance|\n",
      "|Large religious f...| finance|\n",
      "|Lower class large...| finance|\n",
      "|Lower class large...| finance|\n",
      "|     Family starters| finance|\n",
      "|       Stable family| finance|\n",
      "|Modern, complete ...| finance|\n",
      "|Lower class large...| finance|\n",
      "|        Mixed rurals| finance|\n",
      "|    Young and rising| finance|\n",
      "|Lower class large...| finance|\n",
      "|Traditional families| finance|\n",
      "|Mixed apartment d...| finance|\n",
      "|Young all america...| finance|\n",
      "|Low income catholics| finance|\n",
      "+--------------------+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select('Customer_subtype','constant').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "### UDFs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 151,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----+\n",
      "|    Avg_age|count|\n",
      "+-----------+-----+\n",
      "|70-80 years|    8|\n",
      "|50-60 years|  373|\n",
      "|30-40 years|  496|\n",
      "|20-30 years|   31|\n",
      "|60-70 years|   64|\n",
      "|40-50 years| 1028|\n",
      "+-----------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupby(\"Avg_age\").count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 157,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create a function to assign categories\n",
    "def age_category(age):\n",
    "    if age  == '20-30 years':\n",
    "        return 'Young'\n",
    "    elif age== '30-40 years':\n",
    "        return 'Mid Aged' \n",
    "    elif ((age== '40-50 years') or (age== '50-60 years')) :\n",
    "        return 'Old'\n",
    "    else:\n",
    "        return 'Very Old'\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 158,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create age category udf \n",
    "age_udf=udf(age_category,StringType())\n",
    "#create the bucket column by applying udf\n",
    "df=df.withColumn('age_category',age_udf(df['Avg_age']))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 159,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+------------+\n",
      "|    Avg_age|age_category|\n",
      "+-----------+------------+\n",
      "|30-40 years|    Mid Aged|\n",
      "|30-40 years|    Mid Aged|\n",
      "|30-40 years|    Mid Aged|\n",
      "|40-50 years|         Old|\n",
      "|30-40 years|    Mid Aged|\n",
      "|20-30 years|       Young|\n",
      "|30-40 years|    Mid Aged|\n",
      "|40-50 years|         Old|\n",
      "|50-60 years|         Old|\n",
      "|40-50 years|         Old|\n",
      "|40-50 years|         Old|\n",
      "|40-50 years|         Old|\n",
      "|40-50 years|         Old|\n",
      "|40-50 years|         Old|\n",
      "|30-40 years|    Mid Aged|\n",
      "|40-50 years|         Old|\n",
      "|40-50 years|         Old|\n",
      "|40-50 years|         Old|\n",
      "|30-40 years|    Mid Aged|\n",
      "|50-60 years|         Old|\n",
      "+-----------+------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select('Avg_age','age_category').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 160,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-----+\n",
      "|age_category|count|\n",
      "+------------+-----+\n",
      "|    Mid Aged|  496|\n",
      "|    Very Old|   72|\n",
      "|         Old| 1401|\n",
      "|       Young|   31|\n",
      "+------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupby(\"age_category\").count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 267,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----------------+\n",
      "|summary|       Avg_Salary|\n",
      "+-------+-----------------+\n",
      "|  count|             2000|\n",
      "|   mean|     1616908.0835|\n",
      "| stddev|6822647.757312146|\n",
      "|    min|             1361|\n",
      "|    25%|            20315|\n",
      "|    50%|            31421|\n",
      "|    75%|            42949|\n",
      "|    max|         48919896|\n",
      "+-------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select('Avg_Salary').summary().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 268,
   "metadata": {},
   "outputs": [],
   "source": [
    "min_sal=1361\n",
    "max_sal=48919896"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 275,
   "metadata": {},
   "outputs": [],
   "source": [
    "### Pandas udf \n",
    "from pyspark.sql.functions import pandas_udf, PandasUDFType\n",
    "\n",
    "def scaled_salary(salary):\n",
    "    scaled_sal=(salary-min_sal)/(max_sal-min_sal)\n",
    "    return scaled_sal"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 276,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------+----------------+------------------+-----------+---------------------+----------+-----+--------+---------------------+\n",
      "|Customer_subtype          |Number_of_houses|Avg_size_household|Avg_age    |Customer_main_type   |Avg_Salary|label|constant|scaled_salary        |\n",
      "+--------------------------+----------------+------------------+-----------+---------------------+----------+-----+--------+---------------------+\n",
      "|Lower class large families|1               |3                 |30-40 years|Family with grown ups|44905     |0    |finance |8.901329526732557E-4 |\n",
      "|Mixed small town dwellers |1               |2                 |30-40 years|Family with grown ups|37575     |0    |finance |7.40291997705982E-4  |\n",
      "|Mixed small town dwellers |1               |2                 |30-40 years|Family with grown ups|27915     |0    |finance |5.42820834679534E-4  |\n",
      "|Modern, complete families |1               |3                 |40-50 years|Average Family       |19504     |0    |finance |3.708819162307293E-4 |\n",
      "|Large family farms        |1               |4                 |30-40 years|Farmers              |34943     |0    |finance |6.864882605335584E-4 |\n",
      "|Young and rising          |1               |2                 |20-30 years|Living well          |13064     |0    |finance |2.3923447421309735E-4|\n",
      "|Large religious families  |2               |3                 |30-40 years|Conservative families|29090     |0    |finance |5.668403602029373E-4 |\n",
      "|Lower class large families|1               |2                 |40-50 years|Family with grown ups|6895      |0    |finance |1.1312685467788436E-4|\n",
      "|Lower class large families|1               |2                 |50-60 years|Family with grown ups|35497     |0    |finance |6.978132112909759E-4 |\n",
      "|Family starters           |2               |3                 |40-50 years|Average Family       |30800     |0    |finance |6.017964356455073E-4 |\n",
      "+--------------------------+----------------+------------------+-----------+---------------------+----------+-----+--------+---------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "scaling_udf = pandas_udf(scaled_salary, DoubleType())\n",
    "df.withColumn(\"scaled_salary\", scaling_udf(df['Avg_Salary'])).show(10,False)\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Joinging "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 176,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------+-----+\n",
      "|Customer_main_type   |count|\n",
      "+---------------------+-----+\n",
      "|Farmers              |93   |\n",
      "|Career Loners        |15   |\n",
      "|Retired and Religious|202  |\n",
      "|Successful hedonists |194  |\n",
      "|Living well          |178  |\n",
      "|Average Family       |308  |\n",
      "|Cruising Seniors     |60   |\n",
      "|Conservative families|236  |\n",
      "|Driven Growers       |172  |\n",
      "|Family with grown ups|542  |\n",
      "+---------------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupby(\"Customer_main_type\").count().show(50,False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 191,
   "metadata": {},
   "outputs": [],
   "source": [
    "region_data = spark.createDataFrame([('Family with grown ups','PN'),\n",
    "                                    ('Driven Growers','GJ'),\n",
    "                                    ('Conservative families','DD'),\n",
    "                                    ('Cruising Seniors','DL'),\n",
    "                                    ('Average Family ','MN'),\n",
    "                                    ('Living well','KA'),\n",
    "                                    ('Successful hedonists','JH'),\n",
    "                                    ('Retired and Religious','AX'),\n",
    "                                   ('Career Loners','HY'),('Farmers','JH')],schema=StructType().add(\"Customer_main_type\",\"string\").add(\"Region Code\",\"string\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 183,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----------+\n",
      "|  Customer_main_type|Region Code|\n",
      "+--------------------+-----------+\n",
      "|Family with grown...|         PN|\n",
      "|      Driven Growers|         GJ|\n",
      "|Conservative fami...|         DD|\n",
      "|    Cruising Seniors|         DL|\n",
      "|     Average Family |         MN|\n",
      "|         Living well|         KA|\n",
      "|Successful hedonists|         JH|\n",
      "|Retired and Relig...|         AX|\n",
      "|       Career Loners|         HY|\n",
      "|             Farmers|         JH|\n",
      "+--------------------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "region_data.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 194,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "new_df=df.join(region_data,on='Customer_main_type')\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 187,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----+\n",
      "|Region Code|count|\n",
      "+-----------+-----+\n",
      "|JH         |287  |\n",
      "|HY         |15   |\n",
      "|DD         |236  |\n",
      "|DL         |60   |\n",
      "|GJ         |172  |\n",
      "|PN         |542  |\n",
      "|KA         |178  |\n",
      "|AX         |202  |\n",
      "+-----------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "new_df.groupby(\"Region Code\").count().show(50,False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Pivoting "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 199,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----------+-----------+-----------+-----------+-----------+-----------+\n",
      "|  Customer_main_type|20-30 years|30-40 years|40-50 years|50-60 years|60-70 years|70-80 years|\n",
      "+--------------------+-----------+-----------+-----------+-----------+-----------+-----------+\n",
      "|             Farmers|          0|     462027|    2031235|     316206|          0|          0|\n",
      "|       Career Loners|     143998|     176639|      25701|     105193|      32558|          0|\n",
      "|Retired and Relig...|     126350|     336631|    2975266|    1687711|     335357|      61124|\n",
      "|Successful hedonists|      42261|  171278764| 1223362814| 1563071675|  200340129|      15518|\n",
      "|         Living well|     460528|    2965303|    1795405|     331304|          0|          0|\n",
      "|      Average Family|          0|   23682805|    7789464|     412490|     226281|          0|\n",
      "|    Cruising Seniors|          0|      43302|     303601|     529354|     716425|     139538|\n",
      "|Conservative fami...|      69390|    2381485|    3449782|     915954|     146432|          0|\n",
      "|      Driven Growers|          0|    1376260|    3407807|     424272|      83936|          0|\n",
      "|Family with grown...|      16406|    2620378|    9132414|    3295378|     162820|      10496|\n",
      "+--------------------+-----------+-----------+-----------+-----------+-----------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('Customer_main_type').pivot('Avg_age').sum('Avg_salary').fillna(0).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 203,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------+---------+\n",
      "|  Customer_main_type|         0|        1|\n",
      "+--------------------+----------+---------+\n",
      "|             Farmers|   2734832|    74636|\n",
      "|       Career Loners|    484089|        0|\n",
      "|Retired and Relig...|   5328410|   194029|\n",
      "|Successful hedonists|2720381462|437729699|\n",
      "|         Living well|   5453384|    99156|\n",
      "|      Average Family|  26036999|  6074041|\n",
      "|    Cruising Seniors|   1675841|    56379|\n",
      "|Conservative fami...|   6595027|   368016|\n",
      "|      Driven Growers|   4492465|   799810|\n",
      "|Family with grown...|  14394094|   843798|\n",
      "+--------------------+----------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('Customer_main_type').pivot('label').sum('Avg_salary').fillna(0).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Window Operations "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "## Ranking "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 127,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.window import Window\n",
    "from pyspark.sql.functions import udf,rank, col,row_number"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 128,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create a window function to order the relevant column( Avg Salary)\n",
    "win = Window.orderBy(df['Avg_Salary'].desc())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 129,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create a additonal column with row numbers as rank\n",
    "df=df.withColumn('rank', row_number().over(win).alias('rank'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 130,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+----+\n",
      "|    Customer_subtype|Number_of_houses|Avg_size_household|    Avg_age|  Customer_main_type|Avg_Salary|label|rank|\n",
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+----+\n",
      "| High status seniors|               1|                 2|60-70 years|Successful hedonists|  48919896|    0|   1|\n",
      "|High Income, expe...|               1|                 2|50-60 years|Successful hedonists|  48177970|    0|   2|\n",
      "|High Income, expe...|               1|                 2|50-60 years|Successful hedonists|  48069548|    1|   3|\n",
      "|High Income, expe...|               1|                 3|40-50 years|Successful hedonists|  46911924|    0|   4|\n",
      "| High status seniors|               1|                 3|40-50 years|Successful hedonists|  46614009|    0|   5|\n",
      "|High Income, expe...|               1|                 3|30-40 years|Successful hedonists|  45952441|    0|   6|\n",
      "|High Income, expe...|               1|                 3|40-50 years|Successful hedonists|  45864609|    1|   7|\n",
      "| High status seniors|               1|                 1|50-60 years|Successful hedonists|  45592572|    0|   8|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|  45170899|    0|   9|\n",
      "|High Income, expe...|               1|                 3|50-60 years|Successful hedonists|  44843830|    0|  10|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|  43230349|    0|  11|\n",
      "| High status seniors|               1|                 5|40-50 years|Successful hedonists|  43181830|    0|  12|\n",
      "| High status seniors|               1|                 1|30-40 years|Successful hedonists|  42631926|    0|  13|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|  41919020|    0|  14|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|  41699271|    0|  15|\n",
      "| High status seniors|               1|                 2|40-50 years|Successful hedonists|  41398953|    0|  16|\n",
      "|High Income, expe...|               1|                 3|40-50 years|Successful hedonists|  41269615|    1|  17|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|  41192397|    1|  18|\n",
      "| High status seniors|               1|                 2|40-50 years|Successful hedonists|  40564335|    0|  19|\n",
      "| High status seniors|               1|                 2|50-60 years|Successful hedonists|  40453887|    0|  20|\n",
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 133,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Ranking groupwise \n",
    "#create a window function to order the relevant column( Avg Salary)\n",
    "win_1 = Window.partitionBy(\"Customer_subtype\").orderBy(df['Avg_Salary'].desc())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 141,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create a additonal column with row numbers as rank\n",
    "df=df.withColumn('rank', row_number().over(win_1).alias('rank'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 143,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+\n",
      "|rank|count|\n",
      "+----+-----+\n",
      "|   1|   39|\n",
      "|   2|   37|\n",
      "|   3|   36|\n",
      "|   4|   36|\n",
      "|   5|   34|\n",
      "|   6|   34|\n",
      "|   7|   32|\n",
      "|   8|   31|\n",
      "|   9|   31|\n",
      "|  10|   31|\n",
      "|  11|   31|\n",
      "|  12|   31|\n",
      "|  13|   31|\n",
      "|  14|   31|\n",
      "|  15|   31|\n",
      "|  16|   30|\n",
      "|  17|   30|\n",
      "|  18|   27|\n",
      "|  19|   27|\n",
      "|  20|   27|\n",
      "+----+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('rank').count().orderBy('rank').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 145,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+----+\n",
      "|    Customer_subtype|Number_of_houses|Avg_size_household|    Avg_age|  Customer_main_type|Avg_Salary|label|rank|\n",
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+----+\n",
      "|Large family, emp...|               2|                 3|30-40 years|Family with grown...|     49418|    0|   1|\n",
      "|Large family, emp...|               1|                 4|40-50 years|Family with grown...|     48390|    0|   2|\n",
      "|Large family, emp...|               1|                 3|40-50 years|Family with grown...|     48272|    0|   3|\n",
      "|Religious elderly...|               1|                 2|50-60 years|Retired and Relig...|     49564|    0|   1|\n",
      "|Religious elderly...|               1|                 2|60-70 years|Retired and Relig...|     49373|    0|   2|\n",
      "|Religious elderly...|               1|                 1|70-80 years|Retired and Relig...|     49059|    0|   3|\n",
      "|Large religious f...|               1|                 3|40-50 years|Conservative fami...|     49547|    0|   1|\n",
      "|Large religious f...|               1|                 2|60-70 years|Conservative fami...|     49268|    0|   2|\n",
      "|Large religious f...|               1|                 3|40-50 years|Conservative fami...|     49040|    0|   3|\n",
      "|Modern, complete ...|               1|                 3|30-40 years|      Average Family|     49729|    0|   1|\n",
      "|Modern, complete ...|               1|                 4|30-40 years|      Average Family|     48801|    0|   2|\n",
      "|Modern, complete ...|               1|                 4|40-50 years|      Average Family|     48694|    0|   3|\n",
      "|    Village families|               1|                 2|50-60 years|Family with grown...|     49575|    0|   1|\n",
      "|    Village families|               1|                 2|30-40 years|Family with grown...|     49097|    0|   2|\n",
      "|    Village families|               1|                 3|40-50 years|Family with grown...|     48108|    0|   3|\n",
      "|Young all america...|               1|                 3|40-50 years|      Average Family|     48433|    0|   1|\n",
      "|Young all america...|               2|                 4|40-50 years|      Average Family|     47538|    0|   2|\n",
      "|Young all america...|               1|                 4|30-40 years|      Average Family|     47393|    0|   3|\n",
      "|Young urban have-...|               2|                 2|50-60 years|         Living well|     31511|    0|   1|\n",
      "|Young urban have-...|               1|                 2|30-40 years|         Living well|     26788|    0|   2|\n",
      "+--------------------+----------------+------------------+-----------+--------------------+----------+-----+----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# filter top 3 customers from every group\n",
    "df.filter(col('rank') < 4).show()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.3"
  },
  "name": "Data Wrangling using DataFrames",
  "notebookId": 1751261886603334
 },
 "nbformat": 4,
 "nbformat_minor": 1
}
